package com.learn.kafka.config;

import com.google.common.collect.Maps;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.kafka.transaction.KafkaTransactionManager;

import java.util.Map;

/**
 * @author: learn.kafka
 */
@Configuration
public class KafkaConfig {

    @Bean
    public KafkaAdmin admin(KafkaProperties properties){

        // 1. 创建管理员配置 Map
        Map<String, Object> adminConfig = Maps.newHashMap();

        // 2. 至少需要配置 bootstrap.servers（从 KafkaProperties 中获取）
        adminConfig.put("bootstrap.servers", properties.getBootstrapServers());
        if (!adminConfig.containsKey(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)) {
            adminConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        }
        if (!adminConfig.containsKey(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG)) {
            adminConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        }
        // 3. 可选：添加其他管理员配置（如超时时间等）
        adminConfig.put("request.timeout.ms", 5000); // 管理员操作超时时间

        // 4. 用配置 Map 初始化 KafkaAdmin
        KafkaAdmin admin = new KafkaAdmin(adminConfig);
        admin.setFatalIfBrokerNotAvailable(true);
        return admin;
    }
    @Bean
    public NewTopic topic2() {
        return TopicBuilder.name("topic-kl").partitions(1)
                .replicas(1)
                .build();
    }

    // 1. 配置支持事务的生产者工厂
    @Bean
    public ProducerFactory<String, Object> producerFactory(KafkaProperties properties) {
        // 获取基础配置
        Map<String, Object> configProps = Maps.newHashMap();
        configProps.put("bootstrap.servers", properties.getBootstrapServers());
        if (!configProps.containsKey(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)) {
            configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        }
        if (!configProps.containsKey(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG)) {
            configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        }
        // 必须配置：启用事务（关键）
        configProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 幂等性是事务的前提
        // 事务ID（唯一标识）
        configProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-id-" + System.currentTimeMillis());
        configProps.put(ProducerConfig.ACKS_CONFIG, "all"); // 事务要求acks=all

        DefaultKafkaProducerFactory<String, Object> factory = new DefaultKafkaProducerFactory<>(configProps);
        factory.setTransactionIdPrefix("tx-prefix-"); // 设置事务ID前缀（可选）
        return factory;
    }
    // 2. 配置支持事务的KafkaTemplate
    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }

    // 3. 配置事务管理器（可选，若需要与Spring事务管理器集成）
    @Bean
    public KafkaTransactionManager<String, Object> transactionManager(ProducerFactory<String, Object> producerFactory) {
        return new KafkaTransactionManager<>(producerFactory);
    }
}
